package SourceMap;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.util.StringUtils;

import java.text.SimpleDateFormat;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

public class ProSourceMap implements MapFunction<String, Tuple4<String,String,Double,Double>> {
    @Override
    public Tuple4<String,String,Double,Double> map(String s) throws Exception {

        if(StringUtils.isNullOrWhitespaceOnly(s)) {
            System.out.println("invalid line");
            return null;
        }
//        SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        String[] array = s.split(",");

        if(null==array || array.length<4) {
            System.out.println("invalid line for array");
            return null;
        }
        //记录省份
        String regex="(?<province>河北|山西|辽宁|吉林|黑龙江|江苏|浙江|安徽|福建|江西|山东|河南|湖北|湖南|广东|海南|四川|贵州|云南|陕西|甘肃|青海|台湾|内蒙古|广西|西藏|宁夏|新疆|北京|天津|上海|重庆|香港|澳门)(?<village>.*)";
        Matcher m= Pattern.compile(regex).matcher(array[5]);
        String province=null;
        while(m.find()){
            province=m.group("province");
        }
        return new Tuple4<>(array[4],province, Double.valueOf(array[8]),Double.valueOf(array[9]));
    }
}
